package com.example.demo.bitmap_test;

import org.roaringbitmap.RoaringBitmap;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import java.io.*;
import java.time.LocalDate;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 记录用户登录工具类
 * */
public class RedisRoaringLoginTracker {

    private final JedisPool jedisPool;

    // 本地缓存：Key=日期, Value=当天的RoaringBitmap
    // 使用ConcurrentHashMap保证线程安全
    private final ConcurrentHashMap<String, RoaringBitmap> localBitmapCache = new ConcurrentHashMap<>();

    // 标识位图脏状态：Key=日期, Value=是否已修改需同步到Redis
    private final ConcurrentHashMap<String, Boolean> dirtyFlags = new ConcurrentHashMap<>();

    public RedisRoaringLoginTracker(String host, int port) {
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        this.jedisPool = new JedisPool(poolConfig, host, port);
    }

    /**
     * 记录用户登录（核心方法）
     * 1. 先操作本地内存中的RoaringBitmap
     * 2. 标记为脏数据，允许异步刷回Redis
     */
    public void recordUserLogin(int userId) {
        String key = getTodayKey();
        // computeIfAbsent是线程安全的，确保只会创建一个RoaringBitmap实例
        RoaringBitmap bitmap = localBitmapCache.computeIfAbsent(key, k -> {
            // 尝试先从Redis反序列化加载已有数据
            RoaringBitmap loaded = loadFromRedis(k);
            return loaded != null ? loaded : new RoaringBitmap();
        });

        // 操作本地位图
        bitmap.add(userId);
        // 标记该日期的位图为“脏”，需要同步到Redis
        dirtyFlags.put(key, true);
    }

    /**
     * 检查用户是否登录（极快，纯内存操作）
     */
    public boolean hasUserLoggedInToday(int userId) {
        String key = getTodayKey();
        RoaringBitmap bitmap = localBitmapCache.get(key);
        return bitmap != null && bitmap.contains(userId);
    }

    /**
     * 获取今日登录人数（极快，纯内存操作）
     */
    public long getTodayLoginCount() {
        String key = getTodayKey();
        RoaringBitmap bitmap = localBitmapCache.get(key);
        return bitmap != null ? bitmap.getCardinality() : 0;
    }

    /**
     * 强制将指定日期的位图同步到Redis（持久化）
     */
    public void flushToRedis(String date) {
        String key = "login:rb:" + date;
        RoaringBitmap bitmap = localBitmapCache.get(key);
        if (bitmap == null) {
            return; // 无数据可同步
        }

        try (Jedis jedis = jedisPool.getResource()) {
            byte[] serializedData = serializeBitmap(bitmap);
            jedis.set(key.getBytes(), serializedData);
            // 设置过期时间（例如7天）
            jedis.expire(key, 60 * 60 * 24 * 7);
            // 清除脏标记
            dirtyFlags.put(key, false);
            System.out.println("已同步位图到Redis: " + key + ", 大小: " + serializedData.length + " 字节");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 从Redis加载位图到本地内存
     */
    private RoaringBitmap loadFromRedis(String date) {
        String key = "login:rb:" + date;
        try (Jedis jedis = jedisPool.getResource()) {
            byte[] data = jedis.get(key.getBytes());
            if (data != null && data.length > 0) {
                RoaringBitmap bitmap = deserializeBitmap(data);
                System.out.println("从Redis加载位图: " + key + ", 用户数: " + bitmap.getCardinality());
                return bitmap;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 序列化 RoaringBitmap 到 byte[]
     */
    private byte[] serializeBitmap(RoaringBitmap bitmap) throws IOException {
        try (ByteArrayOutputStream bos = new ByteArrayOutputStream();
             DataOutputStream dos = new DataOutputStream(bos)) {
            bitmap.serialize(dos);
            return bos.toByteArray();
        }
    }

    /**
     * 反序列化 byte[] 到 RoaringBitmap
     */
    private RoaringBitmap deserializeBitmap(byte[] data) throws IOException {
        RoaringBitmap bitmap = new RoaringBitmap();
        try (ByteArrayInputStream bis = new ByteArrayInputStream(data);
             DataInputStream dis = new DataInputStream(bis)) {
            bitmap.deserialize(dis);
        }
        return bitmap;
    }

    /**
     * 复杂的交集操作示例（纯内存，高性能）
     */
    public RoaringBitmap getUsersLoggedInBothDays(String date1, String date2) {
        // 确保两个日期的位图都已加载到内存
        RoaringBitmap bitmap1 = localBitmapCache.computeIfAbsent("login:rb:" + date1, k -> loadFromRedis(date1) != null ? loadFromRedis(date1) : new RoaringBitmap());
        RoaringBitmap bitmap2 = localBitmapCache.computeIfAbsent("login:rb:" + date2, k -> loadFromRedis(date2) != null ? loadFromRedis(date2) : new RoaringBitmap());

        // 在内存中执行高效的交集操作
        return RoaringBitmap.and(bitmap1, bitmap2);
    }

    /**
     * 定时任务：将脏数据刷回Redis
     */
    public void startSyncTask() {
        new Thread(() -> {
            while (true) {
                try {
                    Thread.sleep(30000); // 每30秒同步一次
                    for (String key : dirtyFlags.keySet()) {
                        if (dirtyFlags.getOrDefault(key, false)) {
                            flushToRedis(key.replace("login:rb:", ""));
                        }
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }, "Redis-Sync-Thread").start();
    }

    private String getTodayKey() {
        return "login:rb:" + LocalDate.now().toString();
    }

    public void close() {
        // 关闭前同步所有脏数据
        for (String key : dirtyFlags.keySet()) {
            if (dirtyFlags.getOrDefault(key, false)) {
                flushToRedis(key.replace("login:rb:", ""));
            }
        }
        if (jedisPool != null) {
            jedisPool.close();
        }
    }

    // 测试
    public static void main(String[] args) throws InterruptedException {
        RedisRoaringLoginTracker tracker = new RedisRoaringLoginTracker("localhost", 6379);
        tracker.startSyncTask(); // 启动后台同步线程

        // 模拟操作
        tracker.recordUserLogin(1);
        tracker.recordUserLogin(1000000); // 稀疏大ID
        tracker.recordUserLogin(999999);

        System.out.println("登录用户数: " + tracker.getTodayLoginCount());
        System.out.println("用户1已登录: " + tracker.hasUserLoggedInToday(1));

        // 等待异步同步
        Thread.sleep(1000);
        tracker.flushToRedis(LocalDate.now().toString()); // 也可以手动强制同步

        tracker.close();
    }
}
